package cn.zd.demo.flink.kafka2db.sink;

import cn.zd.demo.flink.kafka2db.dto.KafkaMsgDto;
import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.*;

public class MysqlSink implements SinkFunction<KafkaMsgDto> {

    private final Connection connection;
    private final PreparedStatement statement;

    public MysqlSink() {
        String url = "jdbc:mysql://localhost:3306/kafka2mysql?autoReconnect=true&useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8&useSSL=false";
        try {
            connection = DriverManager.getConnection(url, "root", "root");
            statement = connection.prepareStatement("INSERT INTO tab_data1 (TOPIC, IN_DATETIME,CTEXT) VALUES (?, ?, ?)");
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void invoke(KafkaMsgDto value, Context context) throws Exception {

        //直接执行更新语句
        statement.setString(1, value.getTopic());
        statement.setDate(2, new Date(System.currentTimeMillis()));
        statement.setString(2, value.getCtext());
        statement.execute();
    }

    @Override
    public void writeWatermark(Watermark watermark) throws Exception {
        SinkFunction.super.writeWatermark(watermark);
    }

    @Override
    public void finish() throws Exception {
        SinkFunction.super.finish();
        if(connection != null) {
            connection.close();
        }
    }
}
